package com.ivyft.ftpserver.hdfs;

import org.apache.ftpserver.filesystem.nativefs.impl.NameEqualsFileFilter;
import org.apache.ftpserver.ftplet.FtpFile;
import org.apache.ftpserver.ftplet.User;
import org.apache.ftpserver.usermanager.impl.WriteRequest;
import org.apache.hadoop.fs.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.util.*;

/**
 * <pre>
 *
 * Created by IntelliJ IDEA.
 * User: zhenqin
 * Date: 15/10/29
 * Time: 12:53
 * To change this template use File | Settings | File Templates.
 *
 * </pre>
 *
 * @author zhenqin
 */
public class HdfsFtpFile implements FtpFile {

    private final Logger LOG = LoggerFactory.getLogger(HdfsFtpFile.class);
    private String fileName;
    private Path file;
    private User user;


    private final FileSystem fs;


    public HdfsFtpFile(String fileName, FileSystem fs, Path path, User user) {
        this.fs = fs;
        this.fileName = fileName;
        this.file = path;
        this.user = user;
        if (fileName == null) {
            throw new IllegalArgumentException("fileName can not be null");
        }
        if (file == null) {
            throw new IllegalArgumentException("file can not be null");
        }

        if (fileName.length() == 0)
            throw new IllegalArgumentException("fileName can not be empty");
        if (fileName.charAt(0) != '/') {
            throw new IllegalArgumentException("fileName must be an absolut path");
        }
    }


    @Override
    public String getAbsolutePath() {
        String fullName = this.fileName;
        int filelen = fullName.length();
        if ((filelen != 1) && (fullName.charAt(filelen - 1) == '/')) {
            fullName = fullName.substring(0, filelen - 1);
        }

        return fullName;
    }


    @Override
    public String getName() {
        if (this.fileName.equals("/")) {
            return "/";
        }

        String shortName = this.fileName;
        int filelen = this.fileName.length();
        if (shortName.charAt(filelen - 1) == '/') {
            shortName = shortName.substring(0, filelen - 1);
        }

        int slashIndex = shortName.lastIndexOf('/');
        if (slashIndex != -1) {
            shortName = shortName.substring(slashIndex + 1);
        }
        return shortName;
    }


    @Override
    public boolean isHidden() {
        return false;
    }


    @Override
    public boolean isDirectory() {
        try {
            return this.fs.isDirectory(file);
        } catch (IOException e) {
            return false;
        }
    }


    @Override
    public boolean isFile() {
        try {
            return this.fs.isFile(file);
        } catch (IOException e) {
            return true;
        }
    }


    @Override
    public boolean doesExist() {
        try {
            return this.fs.exists(file);
        } catch (IOException e) {
            return false;
        }
    }


    @Override
    public long getSize() {
        try {
            return this.fs.getFileStatus(file).getLen();
        } catch (IOException e) {
            return 0L;
        }
    }


    @Override
    public String getOwnerName() {
        try {
            return this.fs.getFileStatus(file).getOwner();
        } catch (IOException e) {
            return "user";
        }

    }


    @Override
    public String getGroupName() {
        try {
            return this.fs.getFileStatus(file).getGroup();
        } catch (IOException e) {
            return "group";
        }

    }


    @Override
    public int getLinkCount() {
        try {
            return this.fs.isDirectory(file) ? 3 : 1;
        } catch (IOException e) {
            return 1;
        }
    }


    @Override
    public long getLastModified() {
        try {
            return this.fs.getFileStatus(file).getModificationTime();
        } catch (IOException e) {
            return System.currentTimeMillis();
        }
    }


    @Override
    public boolean setLastModified(long time) {
        return true;
    }


    @Override
    public boolean isReadable() {
        try {
            if (this.fs.exists(file)) {
                boolean write = fs.getFileStatus(file).getAccessTime() > 0;
                this.LOG.debug("Checking can write: " + write);
                return write;
            }
        } catch (IOException e) {
            return false;
        }

        return false;
    }


    @Override
    public boolean isWritable() {
        this.LOG.debug("Checking authorization for " + getAbsolutePath());
        if (this.user.authorize(new WriteRequest(getAbsolutePath())) == null) {
            this.LOG.debug("Not authorized");
            return false;
        }

        this.LOG.debug("Checking if file exists");
        try {
            if (this.fs.exists(file)) {
                boolean write = fs.getFileStatus(file).getAccessTime() > 0;
                this.LOG.debug("Checking can write: " + write);
                return write;
            }
        } catch (IOException e) {
            return false;
        }

        this.LOG.debug("Authorized");
        return true;
    }


    @Override
    public boolean isRemovable() {
        if ("/".equals(this.fileName)) {
            return false;
        }

        String fullName = getAbsolutePath();

        if (this.user.authorize(new WriteRequest(fullName)) == null) {
            return false;
        }

        int indexOfSlash = fullName.lastIndexOf('/');
        String parentFullName;
        if (indexOfSlash == 0)
            parentFullName = "/";
        else {
            parentFullName = fullName.substring(0, indexOfSlash);
        }

        HdfsFtpFile parentObject = new HdfsFtpFile(parentFullName, fs,  this.file.getParent(), this.user);

        return parentObject.isWritable();
    }


    @Override
    public boolean delete() {
        boolean retVal = false;
        if (isRemovable()) {
            try {
                retVal = this.fs.delete(file, true);
            } catch (IOException e) {
                return false;
            }
        }
        return retVal;
    }


    @Override
    public boolean move(FtpFile dest) {
        boolean retVal = false;
        if ((dest.isWritable()) && (isReadable())) {
            Path destFile = ((HdfsFtpFile) dest).file;

            try {
                if (fs.exists(destFile)) {
                    retVal = false;
                } else {
                    retVal = this.fs.rename(file, destFile);
                }
            } catch (IOException e) {
                return false;
            }
        }

        return retVal;
    }


    @Override
    public boolean mkdir() {
        boolean retVal = false;
        if (isWritable()) {
            try {
                retVal = this.fs.mkdirs(file);
            } catch (IOException e) {
                return false;
            }
        }
        return retVal;
    }


    @Override
    public List<FtpFile> listFiles() {
        try {
            if (!this.fs.isDirectory(file)) {
                return null;
            }
        } catch (IOException e) {
            return null;
        }

        FileStatus[] fileStatuses;
        try {
            fileStatuses = this.fs.listStatus(file);
        } catch (IOException e) {
            return null;
        }
        //File[] files = (File[]) fileStatuses;
        if (fileStatuses == null || fileStatuses.length == 0) {
            return null;
        }

        Arrays.sort(fileStatuses, new Comparator<FileStatus>() {
            public int compare(FileStatus f1, FileStatus f2) {
                return f1.getPath().getName().compareTo(f2.getPath().getName());
            }
        });
        String virtualFileStr = getAbsolutePath();
        if (virtualFileStr.charAt(virtualFileStr.length() - 1) != '/') {
            virtualFileStr = virtualFileStr + '/';
        }

        FtpFile[] virtualFiles = new FtpFile[fileStatuses.length];
        for (int i = 0; i < fileStatuses.length; i++) {
            FileStatus fileObj = fileStatuses[i];
            String fileName = virtualFileStr + fileObj.getPath().getName();
            virtualFiles[i] = new HdfsFtpFile(fileName, fs, fileObj.getPath(), this.user);
        }

        return Collections.unmodifiableList(Arrays.asList(virtualFiles));
    }


    @Override
    public OutputStream createOutputStream(long offset)
            throws IOException {
        if (!isWritable()) {
            throw new IOException("No write permission : " + this.file.getName());
        }

        if(offset > 0) {
            throw new UnsupportedOperationException("hdfs file can't support access writing, offset = " + offset);
        }

        return fs.create(this.file, true);
    }


    @Override
    public InputStream createInputStream(long offset)
            throws IOException {
        if (!isReadable()) {
            throw new IOException("No read permission : " + this.file.getName());
        }

        FSDataInputStream inputStream = fs.open(file);
        inputStream.seek(offset);
        return inputStream;

        /*
        final RandomAccessFile raf = new RandomAccessFile(this.file, "r");
        raf.seek(offset);

        return new FileInputStream(raf.getFD())
        {
            public void close() throws IOException {
                super.close();
                raf.close();
            }
        };*/
    }



    public Path getPhysicalFile() {
        return this.file;
    }


    public static final String normalizeSeparateChar(String pathName) {
        String normalizedPathName = pathName.replace(File.separatorChar, '/');
        normalizedPathName = normalizedPathName.replace('\\', '/');
        return normalizedPathName;
    }

    public static final String getPhysicalName(String rootDir, String currDir, String fileName) {
        return getPhysicalName(rootDir, currDir, fileName, false);
    }

    public static final String getPhysicalName(String rootDir, String currDir, String fileName, boolean caseInsensitive) {
        String normalizedRootDir = normalizeSeparateChar(rootDir);
        if (normalizedRootDir.charAt(normalizedRootDir.length() - 1) != '/') {
            normalizedRootDir = normalizedRootDir + '/';
        }

        String normalizedFileName = normalizeSeparateChar(fileName);

        String normalizedCurrDir = currDir;
        String resArg;
        if (normalizedFileName.charAt(0) != '/') {
            if (normalizedCurrDir == null) {
                normalizedCurrDir = "/";
            }
            if (normalizedCurrDir.length() == 0) {
                normalizedCurrDir = "/";
            }

            normalizedCurrDir = normalizeSeparateChar(normalizedCurrDir);

            if (normalizedCurrDir.charAt(0) != '/') {
                normalizedCurrDir = '/' + normalizedCurrDir;
            }
            if (normalizedCurrDir.charAt(normalizedCurrDir.length() - 1) != '/') {
                normalizedCurrDir = normalizedCurrDir + '/';
            }

            resArg = normalizedRootDir + normalizedCurrDir.substring(1);
        } else {
            resArg = normalizedRootDir;
        }

        if (resArg.charAt(resArg.length() - 1) == '/') {
            resArg = resArg.substring(0, resArg.length() - 1);
        }

        StringTokenizer st = new StringTokenizer(normalizedFileName, "/");
        while (st.hasMoreTokens()) {
            String tok = st.nextToken();

            if (!tok.equals(".")) {
                if (tok.equals("..")) {
                    if (resArg.startsWith(normalizedRootDir)) {
                        int slashIndex = resArg.lastIndexOf('/');
                        if (slashIndex != -1) {
                            resArg = resArg.substring(0, slashIndex);
                        }

                    }

                } else if (tok.equals("~")) {
                    resArg = normalizedRootDir.substring(0, normalizedRootDir.length() - 1);
                } else {
                    if (caseInsensitive) {
                        File[] matches = new File(resArg).listFiles(new NameEqualsFileFilter(tok, true));

                        if ((matches != null) && (matches.length > 0)) {
                            tok = matches[0].getName();
                        }
                    }

                    resArg = resArg + '/' + tok;
                }
            }
        }
        if (resArg.length() + 1 == normalizedRootDir.length()) {
            resArg = resArg + '/';
        }

        if (!resArg.regionMatches(0, normalizedRootDir, 0, normalizedRootDir.length())) {
            resArg = normalizedRootDir;
        }

        return resArg;
    }


    @Override
    public boolean equals(Object obj) {
        if ((obj instanceof HdfsFtpFile)) {
            String thisCanonicalPath;
            String otherCanonicalPath;
            try {
                thisCanonicalPath = this.file.toString();
                otherCanonicalPath = ((HdfsFtpFile) obj).file.toString();
            } catch (Exception e) {
                throw new RuntimeException("Failed to get the canonical path", e);
            }

            return thisCanonicalPath.equals(otherCanonicalPath);
        }
        return false;
    }


    @Override
    public int hashCode() {
        try {
            return this.file.toString().hashCode();
        } catch (Exception e) {
        }
        return 0;
    }
}
